#include "my_buf.h"
#include "my_buf.h"
#include "my_log.h"
#include "my_crc.h"
#include "my_thread_base.h"

my_buf_data_t *
my_buf_data_alloc(size_t size)
{
    my_buf_data_t *buf_data;
    buf_data = my_thread_calloc(sizeof(my_buf_data_t));
    buf_data->refer = 1;
    buf_data->data = my_thread_malloc(size);
    buf_data->size = size;
    return buf_data;
}


my_buf_data_t *
my_buf_data_aligned_alloc(size_t size)
{
    my_buf_data_t *buf_data;
    buf_data = my_thread_calloc(sizeof(my_buf_data_t));
    buf_data->refer = 1;
    buf_data->data = my_thread_aligned_malloc(size);
    buf_data->size = size;
    buf_data->aligned = 1;
    return buf_data;
}


void
my_buf_data_free(my_buf_data_t *buf_data)
{
    int64_t new_refer;
    new_refer = my_atomic_int64_add_fetch(&buf_data->refer, -1);

    if (new_refer < 0) {
        MY_DEF_LOG_FATAL("new_refer must be greater than or equal to 0");
    }

    if (new_refer != 0) {
        return;
    }

    if (buf_data->aligned) {
        my_thread_aligned_free(buf_data->data);
    } else {
        my_thread_free(buf_data->data);
    }

    my_thread_free(buf_data);
}


void
my_buf_data_hold(my_buf_data_t *buf_data)
{
    int64_t old_refer;
    old_refer = my_atomic_int64_fetch_add(&buf_data->refer, 1);
    if (old_refer < 1) {
        MY_DEF_LOG_FATAL("old_refer must be greater than 0");
    }
}


my_buf_t *
my_buf_alloc(my_buf_data_t *buf_data, char *start, char *end)
{
    assert((start <= end) && 
        (start >= buf_data->data) &&
        (end <= buf_data->data + buf_data->size));

    my_buf_data_hold(buf_data);

    my_buf_t *buf;
    buf = my_thread_calloc(sizeof(my_buf_t));
    buf->refer = 1;
    buf->data = buf_data;
    buf->data_end = buf_data->data + buf_data->size;
    buf->start = start;
    buf->pos = start;
    buf->end = end;

    return buf;
}


my_buf_t *
my_buf_alloc_ext(size_t size)
{
    my_buf_data_t   *buf_data;
    my_buf_t        *buf;

    buf_data = my_buf_data_alloc(size);
    buf = my_buf_alloc(buf_data, buf_data->data, buf_data->data);
    my_buf_data_free(buf_data);

    return buf;
}


my_buf_t *
my_buf_aligned_alloc_ext(size_t size)
{
    my_buf_data_t   *buf_data;
    my_buf_t        *buf;

    buf_data = my_buf_data_aligned_alloc(size);
    buf = my_buf_alloc(buf_data, buf_data->data, buf_data->data);
    my_buf_data_free(buf_data);

    return buf;
}


void
my_buf_free(my_buf_t *buf)
{
    int64_t new_refer;
    new_refer = my_atomic_int64_add_fetch(&buf->refer, -1);
    if (new_refer < 0) {
        MY_DEF_LOG_FATAL("new_refer must be greater than or equal to 0");
    }

    if (new_refer != 0) {
        return;
    }

    my_buf_data_free(buf->data);
    my_thread_free(buf);
}


void
my_buf_hold(my_buf_t *buf)
{
    int64_t old_refer = my_atomic_int64_fetch_add(&buf->refer, 1);
    if (old_refer < 1) {
        MY_DEF_LOG_FATAL("old_refer must be gerater than 0");
    }
}


uint32_t
my_buf_crc32(my_buf_t *buf)
{
    uint32_t crc = 0;
    my_crc32_init(crc);
    my_crc32_update(&crc, (u_char *)buf->pos, (size_t)(buf->end - buf->pos));
    my_crc32_final(crc);
    return crc;
}


my_buf_chain_t *
my_buf_chain_alloc()
{
    my_buf_chain_t *chain;
    chain = my_thread_calloc(sizeof(my_buf_chain_t));
    my_queue_init(&chain->head);
    return chain;
}


void
my_buf_chain_free(my_buf_chain_t *chain)
{
    my_buf_t    *buf;
    my_queue_t  *q;

    while (!my_queue_empty(&chain->head)) {
        q = my_queue_last(&chain->head);
        my_queue_remove(q);
        buf = my_queue_data(q, my_buf_t, queue);
        my_buf_free(buf);
    }

    my_thread_free(chain);
}


void
my_buf_chain_enqueue(my_buf_chain_t *chain, my_buf_t *buf)
{
    my_buf_hold(buf);
    my_queue_insert_head(&chain->head, &buf->queue);
}


my_buf_t *
my_buf_chain_dequeue(my_buf_chain_t *chain)
{
    my_queue_t      *q;
    my_buf_t        *buf;

    if (my_queue_empty(&chain->head)) {
        return NULL;
    }

    q = my_queue_last(&chain->head);
    my_queue_remove(q);
    buf = my_queue_data(q, my_buf_t, queue);

    return buf;
}


void
my_buf_chain_consume(my_buf_chain_t *chain, size_t n)
{
    my_queue_t      *q;
    my_buf_t        *buf;
    size_t          data_len;
    size_t          todo;

    while(!my_queue_empty(&chain->head)) {
        q = my_queue_last(&chain->head);
        buf = my_queue_data(q, my_buf_t, queue);
        data_len = (size_t)(buf->end - buf->pos);

        todo = data_len;
        if (todo > n) {
            todo = n;
        }

        buf->pos += todo;
        n -= todo;
        data_len -= todo;

        if (n == 0) {
            break;
        }

        // do not release buf after consuming buf data.
        // you can add data by modifying last_buf.
        my_queue_remove(q);
        my_buf_free(buf);
    }

    assert(n == 0);
}


void
my_buf_chain_copy(my_buf_chain_t *dest, const my_buf_chain_t *src)
{
    my_buf_t        *buf, *new_buf;
    my_queue_t      *q;

    q = my_queue_last(&src->head);
    while(q != &src->head) {
        buf = my_queue_data(q, my_buf_t, queue);
        new_buf = my_buf_alloc(buf->data, buf->pos, buf->end);
        my_buf_chain_enqueue(dest, new_buf);
        my_buf_free(new_buf);   // must free it
        q = q->prev;
    }
}


void
my_buf_chain_copy_n(my_buf_chain_t *dest, const my_buf_chain_t *src,
    size_t offset, size_t len)
{
    char        *end;
    size_t      data_len;
    my_buf_t    *buf, *new_buf;

    my_queue_t  *q = my_queue_last(&src->head);

    // skip the previous buff
    while (offset > 0 && q != &src->head) {
        buf = my_queue_data(q, my_buf_t, queue);

        end = buf->end;
        data_len = (size_t)(buf->end - buf->pos);

        if (data_len > offset) {
            break;
        }

        offset -= data_len;
        q = q->prev;
    }

    char        *buf_pos;
    while(len > 0 && q != &src->head) {
        buf = my_queue_data(q, my_buf_t, queue);

        buf_pos = buf->pos + offset;
        offset = 0;

        end = buf->end;
        data_len = (size_t)(buf->end - buf_pos);

        if (data_len > len) {
            end = buf_pos + len;
            data_len = len;
        }

        len -= data_len;
        new_buf = my_buf_alloc(buf->data, buf_pos, end);
        my_buf_chain_enqueue(dest, new_buf);
        my_buf_free(new_buf);   // must free it
        q = q->prev;
    }

    assert(len == 0);
}


// @return: last writable buffer.
// defferent from dequeue, no remove queue node.
my_buf_t *
my_buf_chain_last(my_buf_chain_t *chain)
{
    my_queue_t      *q;
    my_buf_t        *buf;

    if (my_queue_empty(&chain->head)) {
        return NULL;
    }

    q = my_queue_head(&chain->head);
    buf = my_queue_data(q, my_buf_t, queue);

    return buf;
}


// @return: first readable buffer.
// no remove queue node
my_buf_t *
my_buf_chain_first(my_buf_chain_t *chain)
{
    my_queue_t      *q;
    my_buf_t        *buf;

    if (my_queue_empty(&chain->head)) {
        return NULL;
    }

    q = my_queue_last(&chain->head);
    buf = my_queue_data(q, my_buf_t, queue);

    return buf;
}


// return next readable buf at the cur buf.
// no remove queue node
my_buf_t *
my_buf_chain_next(my_buf_chain_t *chain, my_buf_t *cur)
{
    if (cur->queue.prev == &chain->head) {
        return NULL;
    }

    return my_queue_data(cur->queue.prev, my_buf_t, queue);
}


size_t
my_buf_chain_total_bytes(my_buf_chain_t *chain)
{
    my_queue_t      *q;
    my_buf_t        *buf;

    size_t          total = 0;
    q = my_queue_last(&chain->head);
    while(q != &chain->head) {
        buf = my_queue_data(q, my_buf_t, queue);
        total += (size_t)(buf->end - buf->pos);
        q = q->prev;
    }

    return total;
}


uint32_t
my_buf_chain_crc32(my_buf_chain_t *chain)
{
    my_queue_t      *q;
    my_buf_t        *buf;

    uint32_t        crc = 0;
    my_crc32_init(crc);

    q = my_queue_last(&chain->head);
    while(q != &chain->head) {
        buf = my_queue_data(q, my_buf_t, queue);
        my_crc32_update(&crc, (u_char *)buf->pos, (size_t)(buf->end - buf->pos));
        q = q->prev;
    }

    my_crc32_final(crc);
    return crc;
}

